package com.shuhe.handler;

import com.alibaba.fastjson.JSONObject;
import com.shuhe.JettyServer;
import com.shuhe.common.HttpClient;
import com.shuhe.common.Tools;
import com.shuhe.domain.dbTodb.DbToDb;
import com.shuhe.domain.mxgc.*;
import com.shuhe.metastore.*;
import org.apache.log4j.Logger;
import org.influxdb.InfluxDB;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.UUID;

import static com.shuhe.common.CommonFunction.*;
import static com.shuhe.common.MyDateUtils.getNowDate;
import static com.shuhe.common.Tools.*;
import static com.shuhe.domain.mxgc.TaskCron.getPushApiConfigs;
import static com.shuhe.handler.CommonHandler.reqExpire;

public class DataInHandler extends HttpServlet {
    private static Logger logger = Logger.getLogger(DataInHandler.class);
    public static String db_type = JettyServer.globalProperties.getProperty("db.type");
    public static String batchId = "";

    @Override
    protected void doOptions(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        resp = setResponse(resp);
    }

    @Override
    protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        try {
            req.setCharacterEncoding("UTF-8");
            resp = setResponse(resp);
            String bodyStr = readAsChars(req);
            JSONObject bodyJson = JSONObject.parseObject(bodyStr);
            MxUserSjjrTaskResponseBody sqlReq = JSONObject.toJavaObject(bodyJson, MxUserSjjrTaskResponseBody.class);
            logger.info(sqlReq.getOperateType());

            //访问日志
            reqExpire(req, bodyJson);
            //（一）数据源管理


            //5、连接测试
            if (sqlReq.getOperateType().equals("datasource_test")) {
                String sqlResp = datasourceTest(sqlReq);
                resp.getWriter().write(sqlResp);
                logger.info(sqlResp);
            }
            //（二）数据接入任务规则
            //8、修改
            if (sqlReq.getOperateType().equals("datatask_modify")) {
                String sqlResp = dataTaskModify(sqlReq, bodyJson);
                resp.getWriter().write(sqlResp);
                logger.info(sqlResp);
            }
            //9、删除
            if (sqlReq.getOperateType().equals("datatask_delete")) {
                String sqlResp = dataTaskDelete(sqlReq);
                resp.getWriter().write(sqlResp);
                logger.info(sqlResp);
            }
            //（三）接入任务管理
            //10、开始任务
            if (sqlReq.getOperateType().equals("datatask_start")) {
                String sqlResp = dataTaskStart(sqlReq);
                resp.getWriter().write(sqlResp);
                logger.info(sqlResp);
            }
            //11、停止任务
            if (sqlReq.getOperateType().equals("datatask_stop")) {
                String sqlResp = dataTaskStop(sqlReq);
                resp.getWriter().write(sqlResp);
                logger.info(sqlResp);
            }
            //12、立即执行
            if (sqlReq.getOperateType().equals("datatask_runimmed")) {
                String sqlResp = dataTaskRunimmed(sqlReq);
                resp.getWriter().write(sqlResp);
                logger.info(sqlResp);
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            logger.info(ex.getMessage());
            resp.getWriter().write(returnFail(ex.getMessage()));
        }
    }


    public static String datasourceTest(MxUserSjjrTaskResponseBody sqlReq) throws Exception {
        Connection metaConn = null;
        CommonResponseBody comm = new CommonResponseBody();

        try {
            String id = sqlReq.getId();
            String sql = "select db_type,ip,source_port,db_name,db_user_name,password from mxgc_user_sjygl " + "where"
                    + " id='" + id + "'";
            metaConn = MetaMysql.connect();
            ResultSet rs = MetaMysql.querySql(metaConn, sql);
            MxUserSjyglResponseBody mxUserSjyglResponseBody = new MxUserSjyglResponseBody();
            while (rs.next()) {
                mxUserSjyglResponseBody.setDbType(rs.getString("db_type"));
                mxUserSjyglResponseBody.setIp(rs.getString("ip"));
                mxUserSjyglResponseBody.setSourcePort(rs.getString("source_port"));
                mxUserSjyglResponseBody.setDbName(rs.getString("db_name"));
                mxUserSjyglResponseBody.setUserNameSource(rs.getString("db_user_name"));
                mxUserSjyglResponseBody.setPassword(rs.getString("password"));
            }
            //influxdb和其他需要分两支
            HashMap<String, String> hashMaps = getConnect(mxUserSjyglResponseBody);
            if (mxUserSjyglResponseBody.getDbType().toLowerCase().equals("influxdb")) {
                InfluxDB influxDB = DbInflux.influxDbBuild(hashMaps.get("url"),
                        mxUserSjyglResponseBody.getUserNameSource(), mxUserSjyglResponseBody.getPassword());
                influxDB.close();
            } else {
                if (mxUserSjyglResponseBody.getDbType().toLowerCase().equals("interface")) {
                    String url = mxUserSjyglResponseBody.getDbName();
                    String resultstr = HttpClient.getHttp(url);
                } else {
                    Connection con = DBSql.connectByAll(hashMaps.get("url"),
                            mxUserSjyglResponseBody.getUserNameSource(), mxUserSjyglResponseBody.getPassword(),
                            hashMaps.get("className"));
                    con.close();
                }
            }
            comm.setSuccess();
            comm.setMsg("测试通过");
            MxUserSjyglResponseBody resp = new MxUserSjyglResponseBody();
            resp.setCommon(comm);
            String respJson = JSONObject.toJSONString(resp);
            return respJson;
        } finally {
            closeDb(metaConn);
        }
    }


    public static String dataTaskModify(MxUserSjjrTaskResponseBody sqlReq, JSONObject bodyJson) throws Exception {
        Connection metaConn = null;
        CommonResponseBody comm = new CommonResponseBody();

        try {
            JSONObject jo = bodyJson.getJSONObject("data");
            String id = jo.getString("id");
            String is_create = jo.getString("is_create");
            String dataSourceId = jo.getString("datasource_id");
            String tableName = jo.getString("table_name");
            String table_comment = jo.getString("table_comment");
            String taskName = jo.getString("task_name");
            String taskStatus = jo.getString("task_status");
            String runCron = jo.getString("run_cron");
            String modifyTime = dateFormat(new Date());
            String cronType = jo.getString("cron_type");
            String timeAddColumn = jo.getString("timeadd_column");
            String timeColumnFormat = jo.getString("timecolumn_format");
            String startTime = jo.getString("start_time");
            String newTime = jo.getString("new_time");
            String destTableName = jo.getString("desttable_name");
            String taskType = jo.getString("task_type");
            String user_name = jo.getString("user_name");
            String timeadd_interval = jo.getString("timeadd_interval");
            String no_task_time_begin = jo.getString("no_task_time_begin");
            String no_task_time_end = jo.getString("no_task_time_end");
            String dir_select = jo.getString("dir_select");

            //啥也不说，直接先删再插了
            String sql = "delete from mxgc_user_task_sjjr where id = '" + id + "'";
            metaConn = MetaMysql.connect();
            MetaMysql.execSql(metaConn, sql);


            sql = "insert into mxgc_user_task_sjjr(id,datasource_id,table_name,task_name,task_status," + "run_cron," + "cron_type,timeadd_column,timecolumn_format,start_time,new_time,modify_time," + "desttable_name," + "task_type, user_name, timeadd_interval, no_task_time_begin, no_task_time_end, " + "mlid)" + "values ('" + id + "','" + dataSourceId + "','" + tableName + "','" + taskName + "','" + taskStatus + "','" + runCron + "','" + cronType + "','" + timeAddColumn + "','" + timeColumnFormat + "','" + startTime + "','" + newTime + "','" + modifyTime + "','" + destTableName + "','" + taskType + "','" + user_name + "','" + timeadd_interval + "','" + no_task_time_begin + "','" + no_task_time_end + "','" + dir_select + "')";
            //是否重新建立目标表
            if (is_create != null && is_create.equals("true") && !db_type.equals("interface")) {
                createDest(user_name, dataSourceId, tableName, table_comment, dir_select, metaConn, destTableName,
                        taskType);
            }
            comm.setSuccess();
            if (comm.getStatus().equals("200")) {
                //规则添加 建表成功后添加规则
                MetaMysql.execSql(metaConn, sql);
            }
            //引用SqlResponseBody 对象，仅为了common格式转换
            SqlResponseBody resp = new SqlResponseBody();
            resp.setCommon(comm);
            String respJson = JSONObject.toJSONString(resp);
            return respJson;
        } finally {
            closeDb(metaConn);
        }
    }

    public static Connection getStgDestConn(String taskType) throws Exception {
        Connection destConn = null;
        if (taskType.equals("stg")) {
            if (db_type.equals("gbase")) {
                destConn = DBJrSql.connect();
            } else {
                destConn = DBSql.connect();
            }
        }
        return destConn;
    }

    public static String returnFail(String reason) {
        MxUserSjyglResponseBody resp = new MxUserSjyglResponseBody();
        CommonResponseBody comm = new CommonResponseBody();
        comm.setFail(reason);
        resp.setCommon(comm);
        String respJson = JSONObject.toJSONString(resp);
        return respJson;
    }

    public static String dataTaskDelete(MxUserSjjrTaskResponseBody sqlReq) throws Exception {
        Connection metaConn = null;
        Connection destConn = null;
        try {
            metaConn = MetaMysql.connect();
            String id = sqlReq.getId();
            String desttableName = sqlReq.getDesttableName();
            String sql = "select 1 from mxgc_user_task_sjjr where table_name = '" + desttableName + "' and task_type "
                    + "= 'push'";
            ResultSet rs = MetaMysql.querySql(metaConn, sql);
            if (rs.next()) {
                return returnFail("相关资源被推送任务使用");
            }
            sql = "select 1 from mxgc_user_table where INSTR(input_tables, '" + desttableName + "') > 0";
            rs = MetaMysql.querySql(metaConn, sql);
            if (rs.next()) {
                return returnFail("相关资源被模型使用");
            }

            String desttableNameNew = "";
            String desttableNameOwner = "";
            if (desttableName.lastIndexOf(".") != -1) {
                desttableNameNew = desttableName.substring(desttableName.lastIndexOf(".") + 1);
                desttableNameOwner = desttableName.substring(0, desttableName.indexOf("."));
            }
            String taskType = "";  //任务分类
            String dataSourceId = "";
            sql = "select * from mxgc_user_task_sjjr where id='" + id + "'";
            rs = MetaMysql.querySql(metaConn, sql);
            if (rs.next()) {
                taskType = rs.getString("task_type");
                dataSourceId = rs.getString("datasource_id");
            }
            //获取数据源的连接 接入是源 推送是目的
            MxUserSjyglResponseBody mxUserSjyglResponseBody = getSjy(metaConn, dataSourceId);
            HashMap<String, String> hashMaps = getConnect(mxUserSjyglResponseBody);



            //目的库表删除
            if (!mxUserSjyglResponseBody.getDbType().equals("interface")) {
                try {
                    //接入的目的库
                    destConn = getStgDestConn(taskType);
                    myDropTable(taskType, desttableNameOwner, desttableNameNew, mxUserSjyglResponseBody, hashMaps,
                            desttableName, destConn);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            //删规则
            sql = "delete from mxgc_user_task_sjjr where id='" + id + "'";
            metaConn = MetaMysql.connect();
            MetaMysql.execSql(metaConn, sql);
            //删日志
            sql = "delete from mxgc_user_task_log_sjjr where task_id='" + id + "'";
            MetaMysql.execSql(metaConn, sql);
            //删除元数据
            sql = "select id from standard_table where table_name = '" + desttableNameNew + "' and table_type = '" + desttableNameOwner + "'";
            rs = MetaMysql.querySql(metaConn, sql);
            if (rs.next()) {
                sql = "delete from standard_table where id = '" + rs.getString("id") + "'";
                MetaMysql.execSql(metaConn, sql);
                sql = "delete from standard_column where table_id = '" + rs.getString("id") + "'";
                MetaMysql.execSql(metaConn, sql);
            }

            MxUserSjyglResponseBody resp = new MxUserSjyglResponseBody();
            CommonResponseBody comm = new CommonResponseBody();
            comm.setSuccess();
            resp.setCommon(comm);
            String respJson = JSONObject.toJSONString(resp);
            return respJson;
        } finally {
            closeDb(metaConn);
        }
    }

    public static void myDropTable(String taskType, String desttableNameOwner, String desttableNameNew,
                                   MxUserSjyglResponseBody mxUserSjyglResponseBody, HashMap<String, String> hashMaps,
                                   String desttableName, Connection destConn) throws Exception {
        String sql = "";
        String cun = "";
        if (taskType.equals("push")) {
            if (destConn == null || destConn.isClosed()) {
                destConn = DBSql.connectByAll(hashMaps.get("url"), mxUserSjyglResponseBody.getUserNameSource(),
                        mxUserSjyglResponseBody.getPassword(), hashMaps.get("className"));
            }
            sql = getSqlTs(mxUserSjyglResponseBody.getDbType(), desttableNameOwner, desttableNameNew);
        } else {
            sql = getSqlTs(db_type, desttableNameOwner, desttableNameNew);
        }
        ResultSet rs = DBSql.querySql(destConn, sql);
        if (rs.next()) {
            cun = rs.getString("cun");
        }
        if (!cun.equals("0")) {
            String sj = dateFormatWfh(new Date());  //去符号日期
            sql = "create table " + desttableName.concat("_").concat(sj) + " as select * from " + desttableName;
            DBSql.execSql(destConn, sql);
            sql = "drop table " + desttableName;
            DBSql.execSql(destConn, sql);
        }
    }

    public static String getSqlTs(String db_type, String owner, String tableName) throws Exception {
        String sql = "";
        if (db_type.equals("mysql") || db_type.startsWith("postgres") || db_type.toLowerCase().equals("gaussdb")) {
            sql = "\tselect  count(*) cun  from information_schema.tables a where table_schema='" + owner.toLowerCase() + "' and" + " a.table_name='" + tableName.toLowerCase() + "' ";
            return sql;
        }
        if (db_type.equals("oracle")) {
            sql = "select COUNT(*) cun from all_tables A where owner='" + owner.toUpperCase() + "' AND A" +
                    ".TABLE_NAME='" + tableName.toUpperCase() + "'";
            return sql;
        }
        if (db_type.equals("gbase")) {
            sql = "select COUNT(*) cun from original.user_tables where table_name = '" + tableName.toUpperCase() + "'";
            return sql;
        }
        return "";
    }

    public static String dataTaskStart(MxUserSjjrTaskResponseBody sqlReq) throws Exception {
        Connection metaConn = null;
        CommonResponseBody comm = new CommonResponseBody();


        try {
            String id = sqlReq.getId();
            String modifyTime = dateFormat(new Date());

            metaConn = MetaMysql.connect();
            String sql = "update mxgc_user_task_sjjr set task_status='run' , modify_time ='" + modifyTime + "' where "
                    + "id='" + id + "'";
            MetaMysql.execSql(metaConn, sql);

            comm.setSuccess();
            MxUserSjyglResponseBody resp = new MxUserSjyglResponseBody();
            comm.setMsg("任务已经开始");
            resp.setCommon(comm);
            String respJson = JSONObject.toJSONString(resp);
            return respJson;
        } finally {
            closeDb(metaConn);
        }
    }

    public static String dataTaskStop(MxUserSjjrTaskResponseBody sqlReq) throws Exception {
        Connection metaConn = null;
        CommonResponseBody comm = new CommonResponseBody();

        try {
            String id = sqlReq.getId();
            String modifyTime = dateFormat(new Date());
            metaConn = MetaMysql.connect();
            String sql = "update mxgc_user_task_sjjr set task_status='stop' , modify_time ='" + modifyTime + "' where"
                    + " id='" + id + "'";
            MetaMysql.execSql(metaConn, sql);
            comm.setSuccess();
            MxUserSjyglResponseBody resp = new MxUserSjyglResponseBody();
            comm.setMsg("任务已经停止");
            resp.setCommon(comm);
            String respJson = JSONObject.toJSONString(resp);
            return respJson;
        } finally {
            closeDb(metaConn);
        }
    }

    public static String dataTaskRunimmed(MxUserSjjrTaskResponseBody sqlReq) throws Exception {
        Connection metaConn = null;
        try {
            metaConn = MetaMysql.connect();
            CommonResponseBody comm = new CommonResponseBody();
            batchId = getUUID();
            String id = sqlReq.getId();
            String sql = "SELECT id,datasource_id,table_name,task_name,task_status,run_cron,cron_type,timeadd_column,"
                    + "timecolumn_format,start_time,new_time,modify_time,desttable_name,task_type,user_name," +
                    "timeadd_interval,no_task_time_begin,no_task_time_end FROM mxgc_user_task_sjjr where id='" + id + "'";
            ArrayList<MxUserSjjrTaskResponseBody> mxUserSjjrTaskResponseBodys =
                    getMxUserSjjrTaskResponseBodys(metaConn, sql);
            if (mxUserSjjrTaskResponseBodys.size() > 0) {
                //获取数据源   接入为源  推送为目标
                MxUserSjyglResponseBody mxUserSjyglResponseBody = getSjy(metaConn,
                        mxUserSjjrTaskResponseBodys.get(0).getDataSourceId());

                //接入、推送合并入口
                logger.info("任务开始" + mxUserSjjrTaskResponseBodys.get(0).getTaskName());
                if (mxUserSjyglResponseBody.getDbType().equals("interface")) {
                    ////接口的单独一支
                    ArrayList<PushApiConfig> pushApiConfigs = new ArrayList<>();
                    sql = "SELECT a.id service_id,a.task_name service_name,a.id service_code,b.platform_code,d" +
                            ".db_name api_server_addr," + "b.api_head,a.user_name,b.token_method,b.token_addr,a" +
                            ".table_name full_table_name " + "FROM mxgc_user_task_sjjr a,push_platform_config b," +
                            "mxgc_user_sjygl d WHERE a.id=b.id " + "AND a.datasource_id=d.id AND a.task_type='push' " + "AND a.task_status='run' AND EXISTS (\n" + "SELECT 1 FROM pz_run_task_server_ts_pc g " + "WHERE g.task_id=a.id ='" + mxUserSjyglResponseBody.getId() + "')";
                    pushApiConfigs = getPushApiConfigs(metaConn, sql);
                    if (pushApiConfigs.size() > 0) {
                        for (int i = 0; i < pushApiConfigs.size(); i++) {
                            HttpPush httpPush = new HttpPush();
                            if (pushApiConfigs.get(i).getPlatformCode().equals("json_001")) {
                                httpPush.HttpPush(pushApiConfigs);
                            } else if (pushApiConfigs.get(i).getPlatformCode().equals("json_002")) {
                                httpPush.httpPush2(pushApiConfigs);  //模拟，无实际意义
                            } else {
                            }
                        }
                    }

                } else {
                    DbToDb.execute(mxUserSjjrTaskResponseBodys.get(0), "hand", batchId);
                }
                logger.info("任务结束" + mxUserSjjrTaskResponseBodys.get(0).getTaskName());
            }
            comm.setRunning();
            MxUserSjyglResponseBody resp = new MxUserSjyglResponseBody();
            resp.setCommon(comm);
            String respJson = JSONObject.toJSONString(resp);
            return respJson;
        } finally {
            closeDb(metaConn);
        }
    }


    public static ArrayList<MxUserSjjrTaskResponseBody> getMxUserSjjrTaskResponseBodys(Connection metaConn,
                                                                                       String sql) throws Exception {
        MxUserSjjrTaskResponseBody mxUserSjjrTaskResponseBody;
        ResultSet rs = MetaMysql.querySql(metaConn, sql);

        ArrayList<MxUserSjjrTaskResponseBody> mxUserSjjrTaskResponseBodys = new ArrayList<>();
        while (rs.next()) {
            mxUserSjjrTaskResponseBody = new MxUserSjjrTaskResponseBody();
            mxUserSjjrTaskResponseBody.setId(rs.getString("id"));
            mxUserSjjrTaskResponseBody.setDataSourceId(rs.getString("datasource_id"));
            mxUserSjjrTaskResponseBody.setTableName(rs.getString("table_name"));
            mxUserSjjrTaskResponseBody.setTaskName(rs.getString("task_name"));
            mxUserSjjrTaskResponseBody.setTaskStatus(rs.getString("task_status"));
            mxUserSjjrTaskResponseBody.setRunCron(rs.getString("run_cron"));
            mxUserSjjrTaskResponseBody.setCronType(rs.getString("cron_type"));
            mxUserSjjrTaskResponseBody.setTimeaddColumn(rs.getString("timeadd_column"));
            mxUserSjjrTaskResponseBody.setTimecolumnFormat(rs.getString("timecolumn_format"));
            mxUserSjjrTaskResponseBody.setStartTime(rs.getString("start_time"));
            mxUserSjjrTaskResponseBody.setNewTime(rs.getString("new_time"));
            mxUserSjjrTaskResponseBody.setModifyTime(rs.getString("modify_time"));
            mxUserSjjrTaskResponseBody.setDesttableName(rs.getString("desttable_name"));
            mxUserSjjrTaskResponseBody.setTaskType(rs.getString("task_type"));
            mxUserSjjrTaskResponseBody.setUserName(rs.getString("user_name"));
            mxUserSjjrTaskResponseBody.setTimeaddInterval(rs.getString("timeadd_interval"));
            mxUserSjjrTaskResponseBody.setNotasktimeBegin(rs.getString("no_task_time_begin"));
            mxUserSjjrTaskResponseBody.setNotasktimeEnd(rs.getString("no_task_time_end"));
            mxUserSjjrTaskResponseBodys.add(mxUserSjjrTaskResponseBody);
        }
        return mxUserSjjrTaskResponseBodys;
    }


    public static String dateFormatWfh(Date date) {
        String dateString = "";
        SimpleDateFormat nyrSfm = new SimpleDateFormat("yyyyMMddHHmmss");
        dateString = nyrSfm.format(date);
        return dateString;
    }


    //字符串转日期
    public static Date strDateFormat(String str) throws Exception {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date date = format.parse(str);
        return date;
    }

    //目标库建表，加两列indb_time,indb_id
    private static void createDest(String user_name, String dataSourceId, String tableName, String table_comment,
                                   String mlid, Connection metaConn, String destTableName, String taskType) throws Exception {

        Connection con = null;  //源端
        InfluxDB influxDB = null;
        Connection destConn = null;
        try {
            //连接源库，获取列名、注释、xh、格式、长度（有倍数关系）  接收各种类型格式
            String destOwner = "";
            String destTableNamenew = "";
            //表名、用户名截取
            destOwner = destTableName.substring(0, destTableName.lastIndexOf("."));
            destTableNamenew = destTableName.substring(destTableName.lastIndexOf(".") + 1);
            //获取连接
            MxUserSjyglResponseBody mxUserSjyglResponseBody = getSjy(metaConn, dataSourceId);
            //接入部分，获取url,className,sql
            HashMap<String, String> hashMaps = getConnect(mxUserSjyglResponseBody);
            String sql = "";
            if (taskType.equals("stg")) {
                //接入时，dataSourceId代表源端
                if (mxUserSjyglResponseBody.getDbType().toLowerCase().equals("influxdb")) {
                    influxDB = DbInflux.influxDbBuild(hashMaps.get("url"),
                            mxUserSjyglResponseBody.getUserNameSource(), mxUserSjyglResponseBody.getPassword());
                } else {
                    con = DBSql.connectByAll(hashMaps.get("url"), mxUserSjyglResponseBody.getUserNameSource(),
                            mxUserSjyglResponseBody.getPassword(), hashMaps.get("className"));
                }
                sql = getSql(mxUserSjyglResponseBody.getDbType(), tableName);
            } else {
                //推送时，配置文件中的db代表源端
                if (JettyServer.globalProperties.getProperty("db.type").equals("gbase")) {
                    con = DBJrSql.connect();
                } else {
                    con = DBSql.connect();
                }
                //目的库为postgres
                sql = getSql(db_type, tableName);
            }
            ArrayList<DbTableBean> dbTableBeans = new ArrayList<>();
            //源数据库字典表（列）转换
            if ((mxUserSjyglResponseBody.getDbType().toLowerCase().equals("influxdb"))) {
                dbTableBeans = Tools.columnAttrChange(influxDB, sql, mxUserSjyglResponseBody.getDbName());
            } else {
                dbTableBeans = Tools.columnAttrChangeQt(mxUserSjyglResponseBody.getDbType(), con, sql, taskType,
                        db_type);
            }
            //存储元数据
            if (taskType.equals("stg")) {
                saveOriginalTable(user_name, destTableName, table_comment, mlid, metaConn, dbTableBeans);
            }
            //拼接create建表语句
            ArrayList<String> comments = new ArrayList<>();
            ArrayList<String> sqlCreateTables = new ArrayList<>();
            sqlCreateTables = concatCreateSql(dbTableBeans, destTableName, mxUserSjyglResponseBody.getDbType(),
                    comments, taskType, db_type);
            if (taskType.equals("stg")) {
                destConn = getStgDestConn(taskType);
                //删表
                myDropTable(taskType, destOwner, destTableNamenew, mxUserSjyglResponseBody, hashMaps, destTableName,
                        destConn);
                for (int j = 0; j < sqlCreateTables.size(); j++) {
                    DBSql.execSql(destConn, sqlCreateTables.get(j));
                }
                //注释维护
                for (int i = 0; i < comments.size(); i++) {
                    DBSql.execSql(destConn, comments.get(i));
                }
            } else {
                destConn = DBSql.connectByAll(hashMaps.get("url"), mxUserSjyglResponseBody.getUserNameSource(),
                        mxUserSjyglResponseBody.getPassword(), hashMaps.get("className"));
                myDropTable(taskType, destOwner, destTableNamenew, mxUserSjyglResponseBody, hashMaps, destTableName,
                        destConn);
                for (int j = 0; j < sqlCreateTables.size(); j++) {
                    DBSql.execSql(destConn, sqlCreateTables.get(j));
                }
            }
        } finally {
            closeDb(destConn);
            closeDb(con);
            closeDbInflux(influxDB);
        }
    }

    public static void saveOriginalTable(String user_name, String tableName, String table_comment, String mlid,
                                         Connection metaConn, ArrayList<DbTableBean> dbTableBeans) throws Exception {
        //存储元数据
        tableName = tableName.split("\\.")[1];
        //将table_name, table_comment, mlid 存储到standard_table
        String deleteSql = "delete from standard_column where table_id = (select id from standard_table where " +
                "table_name = '" + tableName + "')";
        MetaMysql.execSql(metaConn, deleteSql);
        deleteSql = "delete from standard_table where table_name = '" + tableName + "'";
        MetaMysql.execSql(metaConn, deleteSql);
        String table_id = getUUID();
        String insertSql = "insert into standard_table(id,user_name, table_name,table_comment,mlid,update_time, " +
                "table_type) " + "values('" + table_id + "','" + user_name + "','" + tableName + "','" + table_comment + "','" + mlid + "','" + getNowDate() + "', 'original')";
        DBSql.execSql(metaConn, insertSql);

        //将dbTableBeans 存储到standard_column
        for (int i = 0; i < dbTableBeans.size(); i++) {
            String column_size = dbTableBeans.get(i).getCharacter_maximum_length();
            if (column_size == null || column_size.equals("")) {
                column_size = dbTableBeans.get(i).getNumeric_precision() + "," + dbTableBeans.get(i).getNumeric_scale();
            }
            insertSql = "insert into standard_column(id,table_id,column_name,column_comment," + "column_type," +
                    "column_size,column_order) values('" + getUUID() + "','" + table_id + "','" + dbTableBeans.get(i).getColumn_name() + "','" + dbTableBeans.get(i).getColumn_comment() + "','" + dbTableBeans.get(i).getData_type() + "','" + column_size + "','" + dbTableBeans.get(i).getOrdinal_position() + "')";

            DBSql.execSql(metaConn, insertSql);
        }
        //存储元数据
    }


    public static MxUserSjyglResponseBody getSjy(Connection metaConn, String dataSourceId) throws Exception {
        MxUserSjyglResponseBody mxUserSjyglResponseBody = new MxUserSjyglResponseBody();
        String sql =
                "select db_type,ip,source_port,db_name,db_user_name,password from mxgc_user_sjygl a " + "where " + "a"
                        + ".id='" + dataSourceId + "'";
        ResultSet rs = MetaMysql.querySql(metaConn, sql);
        if (rs.next()) {
            mxUserSjyglResponseBody.setDbType(rs.getString("db_type"));
            mxUserSjyglResponseBody.setIp(rs.getString("ip"));
            mxUserSjyglResponseBody.setSourcePort(rs.getString("source_port"));
            mxUserSjyglResponseBody.setDbName(rs.getString("db_name"));
            mxUserSjyglResponseBody.setUserNameSource(rs.getString("db_user_name"));
            mxUserSjyglResponseBody.setPassword(rs.getString("password"));
        }
        return mxUserSjyglResponseBody;
    }


}
